package data.platform.cassandra.infra.persistence.repository;

import com.datastax.oss.driver.api.core.CqlSession;
import data.platform.cassandra.infra.persistence.mapping.RulePointEO;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.GroupedFlux;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Objects;

@ConditionalOnBean(name = "cassandraConfig")
@Repository
@AllArgsConstructor
@Slf4j
public class CassandraRulePointRepository {

    final ReactiveCassandraTemplate reactiveCassandraTemplate;

    final CqlSession cqlSession;

    private static String CREATE_SQL = "INSERT INTO rule_point (rule_id, day, offset, hit_number, hit_data) VALUES (?, ?, ?, ?, ?)";

    @PostConstruct
    public void init() {
        reactiveCassandraTemplate.setApplicationEventPublisher(null);
    }

    public Mono<RulePointEO> insert(RulePointEO entity) {
        return reactiveCassandraTemplate.getReactiveCqlOperations().execute(CREATE_SQL, entity.getRulePointKey().getRuleId()
                , entity.getRulePointKey().getPartition()
                , entity.getRulePointKey().getOffset()
                , entity.getHitNumber()
                , entity.getHitData())
                .map(result -> entity);
    }

    public Flux<RulePointEO> findBySql(Flux<String> sqlFlux) {
        return sqlFlux.flatMap(sql -> reactiveCassandraTemplate.select(sql, RulePointEO.class));
    }

    public Flux<RulePointEO> find(Flux<GroupedFlux<Integer, String>> groups) {
        return groups.flatMap(groupFlux -> findBySql(groupFlux))
                .filter(eo -> Objects.nonNull(eo.getRulePointKey().getRuleId()));
    }

}
